"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : pg_drop_replication_slot函数验证
Create At   : 2022/03/07
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.配置用户，修改pg_hba文件
    4.修改逻辑复制相关参数
    5.创建复制槽
    6.在主机上删除复制槽
    7.在备机上删除复制槽
    8.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.创建成功，修改pg_hba文件成功
    4.修改逻辑复制相关参数成功
    5.创建成功
    6.命名正确的删除成功,命名不符合规范的合理报错
    7.备机不支持,合理报错
    8.资源清理成功，恢复参数成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.standby_root = Node('Standby1Root')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0006')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} &&
            mkdir {self.dir_replication} &&
            cp {self.postgresql_path} {self.postgresql_bak_path} &&
            cp {self.pg_hba_path} {self.pg_hba_bak_path} &&
            chown -R {self.primary_node.ssh_user}:{self.primary_node.ssh_user} {self.dir_replication} &&echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)
        cp_sta_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_sta_msg)
        self.assertEqual("OK", cp_sta_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                                     self.constant.GSGUC_SUCCESS_MSG,
                                                     f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:配置用户,修改pg_hba文件,创建表;expect:成功-----"
        self.logger.info(text)
        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 sha256'
        param_3 = f'host replication {self.primary_node.ssh_user} ' \
            f'::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step4:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                                     self.constant.GSGUC_SUCCESS_MSG,
                                                     f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                                     self.constant.GSGUC_SUCCESS_MSG,
                                                     f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建复制槽;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f"select pg_create_logical_replication_slot" \
            f"('slrp_040_1', 'mppdb_decoding');" \
            f"select pg_create_logical_replication_slot" \
            f"('slrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_" \
            f"040_sslrp', 'mppdb_decoding');"
        create_msg = self.sh_primy.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        sql_cmd = f"select count(*)||'c' from pg_get_replication_slots() " \
            f"where slot_name like 'slrp_040_%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("2c", sql_msg, '执行失败:' + text)

        text = "-----step6:在主机上删除复制槽，" \
               "expect:命名正确的删除成功,命名不符合规范的合理报错-----"
        self.logger.info(text)
        text = "-----step6.1:在主机上删除复制槽，命名包括大写字母" \
               "expect:合理报错-----"
        self.logger.info(text)
        drop_cmd = f"select pg_drop_replication_slot('Slrp_040_1');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertIn("ERROR" and self.constant.invalid_character,
                      drop_msg, '执行失败:' + text)

        text = "-----step6.2:在主机上删除复制槽,命名包含特殊字符*;" \
               "expect:合理报错-----"
        self.logger.info(text)
        drop_cmd = f"select pg_drop_replication_slot('Slrp_040_*');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertIn("ERROR" and self.constant.invalid_character,
                      drop_msg, '执行失败:' + text)

        text = "-----step6.3:创建复制槽,命名长度大于或等于64个字符;" \
               "expect:第一次删除成功，第二次合理报错-----"
        self.logger.info(text)
        drop_cmd = f"select pg_drop_replication_slot" \
            f"('slrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_" \
            f"040_sslrp');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertTrue("ERROR" not in drop_msg and "(1 row)" in drop_msg,
                        '执行失败:' + text)
        drop_cmd = f"select pg_drop_replication_slot" \
            f"('slrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_040_sslrp_" \
            f"040_sslrp_');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertIn("ERROR" and "does not exist",
                      drop_msg, '执行失败:' + text)

        text = "-----step6.4:删除复制槽,命名为空;expect:合理报错-----"
        self.logger.info(text)
        drop_cmd = f"select pg_drop_replication_slot('');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertIn("ERROR:  inputString should not be NULL",
                      drop_msg, '执行失败:' + text)
        sql_cmd = f"select count(*)||'c' from pg_get_replication_slots() " \
            f"where slot_name like 'slrp_040_%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("1c", sql_msg, '执行失败:' + text)

        text = "-----step7:在备机上删除复制槽;expect:备机不支持此函数,合理报错-----"
        self.logger.info(text)
        create_cmd = f"select pg_drop_replication_slot('slrp_040_1');"
        create_msg = self.sh_standby.execut_db_sql(create_cmd,
                                                   dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn("ERROR:  Standby mode doesn't support "
                      "drop logical slot", create_msg, '执行失败:' + text)

    def tearDown(self):        
        text = "-----step8:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step8.1:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f"select pg_drop_replication_slot('slrp_040_1');"
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        self.assertNotIn("ERROR", drop_msg, '执行失败:' + text_1)
        sql_cmd = f"select count(*)||'c' from pg_get_replication_slots() " \
            f"where slot_name like 'slrp_040_%';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("0c", sql_msg, '执行失败:' + text_1)
        
        text_2 = "-----step8.1:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''if [[ -e {self.postgresql_path}
                    && -e {self.postgresql_bak_path} ]];
                    then rm -rf {self.postgresql_path};
                    mv {self.postgresql_bak_path} {self.postgresql_path};
                    echo OK;
                    else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_pri_msg_1 = self.primary_node.sh(restore_cmd_1).result()
        self.logger.info(restore_pri_msg_1)
        restore_sta_msg_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_sta_msg_1)
        restore_cmd_2 = f'''
                    if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
                    then rm -rf {self.pg_hba_path};
                    mv {self.pg_hba_bak_path} {self.pg_hba_path};
                    echo OK;
                    else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_pri_msg_2 = self.primary_node.sh(restore_cmd_2).result()
        self.logger.info(restore_pri_msg_2)
        restore_sta_msg_2 = self.standby_node.sh(restore_cmd_2).result()
        self.logger.info(restore_sta_msg_2)
        self.assertEqual('OK', restore_pri_msg_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_sta_msg_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_pri_msg_2, '执行失败' + text_2)
        self.assertEqual('OK', restore_sta_msg_2, '执行失败' + text_2)
        text_3 = "-----step8.2:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.primary_node.sh(del_cmd)
        self.standby_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
                    then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standy_file_msg = self.standby_node.sh(file_cmd).result()
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
